package com.ljn.data.dao.impl;

import com.ljn.data.dao.ElasticsearchApi;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import javax.annotation.Resource;

import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.MatchAllQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

@Service("elasticsearchApi")
public class ElasticsearchApiImpl implements ElasticsearchApi {

    private static final Logger logger = LoggerFactory.getLogger(ElasticsearchApiImpl.class);


    private static final String EXCEPTION_MESSAGE = "请求出现异常[{}]:{}";

    @Resource
    private RestHighLevelClient client;

    @Override
    public IndexResponse index(IndexRequest request) {
        try {
            return client.index(request);
        } catch (IOException e) {
            logger.error(e.getMessage());
        }
        return null;
    }

    @Override
    public void indexAsync(IndexRequest request) {

        client.indexAsync(request, new ActionListener<IndexResponse>() {
            @Override
            public void onResponse(IndexResponse indexResponse) {
                if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
                    logger.info("CREATED ===>> 索引成功");
                } else if (indexResponse.getResult() == DocWriteResponse.Result.UPDATED) {
                    logger.info("UPDATED ===>> 索引成功");
                }
            }

            @Override
            public void onFailure(Exception e) {
                logger.info("FAILED ===>> 索引失败...\n Cause :: " + e);
            }
        });
    }

    @Override
    public IndexRequest indexRequest(String index, String type, String id, String json) {
        return new IndexRequest(index, type, id).source(json, XContentType.JSON);
    }

    @Override
    public BulkRequest bulkRequest(String index, String type, Map<String, String> jsonMap) {
        BulkRequest bulkRequest = new BulkRequest();
        // 将其他请求绑定到批量请求上，不止是indexRequest，其他查询，获取，删除操作均可以添加到批量请求，统一操作
        for (Map.Entry<String, String> entry : jsonMap.entrySet()) {
            String id = entry.getKey();
            String json = entry.getValue();
            IndexRequest request = this.indexRequest(index, type, id, json);
            request.source(json, XContentType.JSON);
            bulkRequest.add(request);
        }
        return bulkRequest;
    }
    
    @Override
    public void exeUpdate(String index, String type,Map<String, String> map) {
    	BulkRequest bulkRequest = new BulkRequest();
    	  for (Map.Entry<String, String> entry : map.entrySet()) {
    		  String id = entry.getKey();
              String json = entry.getValue();  
              UpdateRequest request = new UpdateRequest(index,type,id);
              request.doc(json, XContentType.JSON);
              bulkRequest.add(request);
    	  }
    	  bulkIndex(bulkRequest);
    }

    @Override
    public void bulkIndex(BulkRequest bulkRequest) {
        try {
            BulkResponse bulkResponse = client.bulk(bulkRequest);
            if (bulkResponse.hasFailures()) {
                logger.error("FAILED ===>> 索引失败！" + bulkResponse.buildFailureMessage());
            }
//            for (BulkItemResponse bulkItemResponse : bulkResponse) {
//                DocWriteResponse itemResponse = bulkItemResponse.getResponse();
//                if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.INDEX
//                        || bulkItemResponse.getOpType() == DocWriteRequest.OpType.CREATE) {
////                	logger.info("CREATED ===>> 索引成功");
//                } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.UPDATE) {
////                	logger.info("UPDATED ===>> 索引成功");
//                } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.DELETE) {
////                	logger.info("DELETE ===>> 索引成功");
//                }
//            }
        } catch (IOException e) {
            logger.error(e.getMessage());
            throw new RuntimeException(e.getMessage());
        }
    }

    @Override
    public void delDocument(String index,String type) {
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.from(0);
        sourceBuilder.size(5000);
        BoolQueryBuilder boolBuilder = QueryBuilders.boolQuery();
        MatchAllQueryBuilder termQueryBuilder = QueryBuilders.matchAllQuery();
        boolBuilder.must(termQueryBuilder);
        sourceBuilder.query(boolBuilder);
        exeDeleteBySearch(index, type, sourceBuilder);
        logger.info("删除文档完毕...");
    }

    @Override
    public void delDocumentById(String index, String type, String id) {
        BulkRequest bulkRequest = new BulkRequest();
        bulkRequest.add(new DeleteRequest(index, type, id));
        this.bulkIndex(bulkRequest);
    }

    @Override
    public void delDocumentByIds(String index, String type, List<String> ids) {
        BulkRequest bulkRequest = new BulkRequest();
        for (String id : ids) {
            bulkRequest.add(new DeleteRequest(index, type, id));
        }
        this.bulkIndex(bulkRequest);
    }

    @Override
    public SearchResponse exeSearch(String index, String type, SearchSourceBuilder sourceBuilder) {
        SearchRequest searchRequest = new SearchRequest(index);
        searchRequest.types(type);
        searchRequest.source(sourceBuilder);
        try {
            return client.search(searchRequest);
        } catch (IOException e) {
            logger.error(e.getMessage());
        }
        return null;
    }

    @Override
    public void exeDeleteBySearch(String index, String type, SearchSourceBuilder sourceBuilder) {
        for (int i = 0; i < 1200; i++) {
            SearchRequest searchRequest = new SearchRequest(index);
            searchRequest.types(type);
            searchRequest.source(sourceBuilder);
            SearchResponse searchResponse;
            Long totalHits = 0L;
            try
            {
                searchResponse = client.search(searchRequest);
                SearchHits hits = searchResponse.getHits();
                totalHits = hits.totalHits;
                if (totalHits == 0) {
                    //所有的已经删除
                    break;
                }
                BulkRequest bulkRequest = new BulkRequest();
                for (SearchHit hit : hits.getHits()) {
                    String id = hit.getId();
                    DeleteRequest deleteRequest = new DeleteRequest(index, type, id);
                    bulkRequest.add(deleteRequest);
                }
                this.bulkIndex(bulkRequest);
                logger.info("删除文档... 循环次数={}", i);

            }
            catch (IOException e)
            {
                logger.error(EXCEPTION_MESSAGE, e);
            }

        }
        logger.info("删除文档完毕...");
    }

}
